package org.hadoop.hive.Utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

/**
 * 作者：Lapper
 * QQ656277239
 * vx:JavaAdmin123
 */

@Component
public class HbaseUtils {
    //每次创建对象都会加载配置
    public Connection getConnection() throws IOException {
        //设置本机客户端访问路径
        System.setProperty("hadoop.home.dir","E:\\hadoop");
        //设置访问身份
        System.setProperty("HADOOP_USER_NAME","root");
        //获取配置对象
        Configuration config = HBaseConfiguration.create();
        //hbase客户端配置zookeeper的集群地址
        config.set("hbase.zookeeper.quorum", "bigdata01,bigdata02,bigdata03");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        return ConnectionFactory.createConnection(config);
    }

    public String createTable(String tableName, String...families){
        try {
            Connection connection = this.getConnection();
            Admin admin=connection.getAdmin();
            TableDescriptorBuilder tableDescriptorBuilder=TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
            for (String family : families) {
                tableDescriptorBuilder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(family));
            }
           admin.createTable(tableDescriptorBuilder.build());
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        return tableName;
    }
    public void put(String tableName,String rowKey,String family,String qualiefierName,String qualiefierValue) throws Exception {
        //获取table对象 操作table
        Connection connection = this.getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        //创建put集合
        List<Put> putList = new ArrayList<>();
        //将行数据放入集合
        putList.add( new Put(Bytes.toBytes(rowKey)).addColumn(Bytes.toBytes(family), Bytes.toBytes(qualiefierName), Bytes.toBytes(qualiefierValue)));
        //提交数据
        table.put(putList);
        table.close();
        connection.close();
        //关闭操作对象
    }
    public void Build(String tableName,String rowKey,String family,String qualiefierName,String qualiefierValue) throws Exception {
        //获取table对象 操作table
        Connection connection=this.getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        //创建put集合
        List<Put> putList = new ArrayList<>();
        //将行数据放入集合
        putList.add( new Put(Bytes.toBytes(rowKey)).addColumn(Bytes.toBytes(family), Bytes.toBytes(qualiefierName), Bytes.toBytes(qualiefierValue)));
        //提交数据
        table.put(putList);
        table.close();
        connection.close();
        //关闭操作对象
    }



    public Map<String, String> get(String tableName, String rowKey) throws Exception {
        //获取table对象 操作table
        Connection connection = this.getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        //行键所对应的所有单元格数据集
        Result result = table.get(new Get(Bytes.toBytes(rowKey)));
        Map<String, String> rowInfo = new HashMap<>();
        //单元格数据集合
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {//cell:每个单元格
            rowInfo.put("timestamp",Long.toString(cell.getTimestamp()));
            rowInfo.put("rowkey",Bytes.toString(CellUtil.cloneRow(cell)));
            rowInfo.put("family",Bytes.toString(CellUtil.cloneFamily(cell)));
            rowInfo.put("qualifier",Bytes.toString(CellUtil.cloneQualifier(cell)));
            rowInfo.put("value",Bytes.toString(CellUtil.cloneValue(cell)));
        }
        table.close();
        connection.close();
        return rowInfo;
    }
    public Map<String, Object> scan(String tableName, Long startTime, Long endTime) throws Exception {
        //获取table对象 操作table
        Connection connection = this.getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        //获取scan对象
        Scan scan = new Scan();
        //得到某一个时间段的数据
        scan.setTimeRange(startTime,endTime);//获取1712734870426L——————————1713251406988L的数据
        //获取全部数据
        Iterator<Result> iterator = table.getScanner(scan).iterator();
        Map<String, Object> allData = new HashMap<>();
        //Map<String,List<Map<String,String>>>
        //遍历
        while (iterator.hasNext()) {
            Map<String, String> qualiefierData = new HashMap<>();
            //获取下一行数据
            Result next = iterator.next();//行数据
            //获取行键
            byte[] row = next.getRow();
            //单元格数据集合
            List<Cell> cells = next.listCells();
            for (Cell cell : cells) {//cell:每个单元格
                qualiefierData.put(Bytes.toString(CellUtil.cloneQualifier(cell)),Bytes.toString(CellUtil.cloneValue(cell)));
            }

            //装载行信息 以rowkey为键  行信息为值
            allData.put(new String(row),qualiefierData);

        }
        //关闭
        table.close();
        connection.close();
        return allData;
    }

    public void delete(String tableName,String rowKey,String family,String qualifier) throws Exception {
        Connection connection = this.getConnection();
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //获取delete对象
        Delete delete = new Delete(rowKey.getBytes());
        //添加删除列
        delete.addColumn(family.getBytes(), qualifier.getBytes());
        //执行删除
        table.delete(delete);
        //关闭
        table.close();
        connection.close();
    }

    public void drop(String tableName) throws Exception {
        //获取操作对象
        Connection connection = this.getConnection();
        Admin admin = connection.getAdmin();
        //删除表之前先停用表
        admin.disableTable(TableName.valueOf(tableName));
        admin.deleteTable(TableName.valueOf(tableName));
        admin.close();
        connection.close();
    }
    public  List<TableName> Showall() throws IOException {
        Connection connection = this.getConnection();
        Admin admin=connection.getAdmin();
        List<TableName> tableNames= Arrays.asList(admin.listTableNames());
        admin.close();
        connection.close();
        return  tableNames;

    }
}
